#include "Service.h"
#include "Sunnet.h"
#include <iostream>
#include <unistd.h>
#include <string.h>

Service::Service() {
    // 同生共死，初始化锁
    pthread_spin_init(&queueLock, PTHREAD_PROCESS_PRIVATE);
    pthread_spin_init(&inGlobalLock, PTHREAD_PROCESS_PRIVATE);
}

Service::~Service() {
    pthread_spin_destroy(&queueLock);
    pthread_spin_destroy(&inGlobalLock);
}

void Service::PushMsg(shared_ptr<BaseMsg> msg) {
    pthread_spin_lock(&queueLock);
    {
        msgQueue.push(msg);
    }
    pthread_spin_unlock(&queueLock);
}

shared_ptr<BaseMsg> Service::PopMsg() {
    shared_ptr<BaseMsg> msg = NULL;

    pthread_spin_lock(&queueLock);
    {
        if (!msgQueue.empty()) {
            msg = msgQueue.front();
            msgQueue.pop();
        }
    }
    pthread_spin_unlock(&queueLock);
    return msg;
}

bool Service::ProcessMsg() {
    shared_ptr<BaseMsg> msg = PopMsg();
    if (msg) {
        OnMsg(msg);
        return true;
    } else {
        return false;
    }
}

void Service::ProcessMsgs(int max) {
    for (int i = 0; i < max; i++) {
        bool succ = ProcessMsg();
        if (!succ) {
            break;
        }
    }
}

void Service::SetInGlobal(bool isIn) {
    pthread_spin_lock(&inGlobalLock);
    {
        inGlobal = isIn;
    }
    pthread_spin_unlock(&inGlobalLock);
    return ;
}

void Service::OnInit() {
    cout << "[" << id << "] OnInit" << endl;
    // Sunnet::instance->Listen(7772, id);

    luaState = luaL_newstate();
    luaL_openlibs(luaState);
    // 注册Sunnet系统API
    LuaAPI::Register(luaState);

    string filename = "../service/" + *type + "/init.lua";
    int isok = luaL_dofile(luaState, filename.data());
    if (isok == 1) { // succ -> 0
        cout << "Run " << filename << " failed: " << lua_tostring(luaState, -1) << endl;
    }

    lua_getglobal(luaState, "OnInit");
    lua_pushinteger(luaState, id); // chunk: ["OnInit", id]
    if (lua_pcall(luaState, 1, 0, 0) != 0) {
        cout << "call lua OnInit failed : " << lua_tostring(luaState, -1) << endl; 
    }
}

void Service::OnMsg(shared_ptr<BaseMsg> msg) {
    if (msg->type == BaseMsg::TYPE::SERVICE) {
        auto serviceMsg = dynamic_pointer_cast<ServiceMsg>(msg);
        OnServiceMsg(serviceMsg);
    } else if (msg->type == BaseMsg::TYPE::SOCKET_ACCEPT) {
        auto socketAcceptMsg = dynamic_pointer_cast<SocketAcceptMsg>(msg);
        OnAcceptMsg(socketAcceptMsg);
    } else if (msg->type == BaseMsg::TYPE::SOCKET_RW) {
        auto socketRWMsg = dynamic_pointer_cast<SocketRWMsg>(msg);
        OnRWMsg(socketRWMsg);
    }
}

void Service::OnExit() {
    cout << "[" << id << "] OnExit" << endl;

    lua_getglobal(luaState, "OnExit");
    if (lua_pcall(luaState, 0, 0, 0) != 0) {
        cout << "call lua OnExit failed : " << lua_tostring(luaState, -1) << endl;
    }

    lua_close(luaState);
}


void Service::OnServiceMsg(shared_ptr<ServiceMsg> serviceMsg) {
    cout << "[" << id << "] OnServiceMsg : " << serviceMsg->buff << endl;

    lua_getglobal(luaState, "OnServiceMsg");
    lua_pushinteger(luaState, serviceMsg->source);
    lua_pushlstring(luaState, serviceMsg->buff.get(), serviceMsg->size);
    if (lua_pcall(luaState, 2, 0, 0) != 0) { // succ -> 0
        cout << "call lua OnServiceMsg failed : " << lua_tostring(luaState, -1) << endl;
    }
}

void Service::OnAcceptMsg(shared_ptr<SocketAcceptMsg> socketAcceptMsg) {
    cout << "OnAcceptMsg" << socketAcceptMsg->clientfd << endl;

    // 新连接绑定ConnWriter
    auto connWriter = make_shared<ConnWriter>();
    connWriter->fd = socketAcceptMsg->clientfd;
    writers.emplace(connWriter->fd, connWriter);

    lua_getglobal(luaState, "OnAcceptMsg");
    lua_pushinteger(luaState, socketAcceptMsg->listenfd);
    lua_pushinteger(luaState, socketAcceptMsg->clientfd);
    if (lua_pcall(luaState, 2, 0, 0) != 0) {
        cout << "call lua OnAcceptMsg failed " << lua_tostring(luaState, -1) << endl; 
    }
}

void Service::OnRWMsg(shared_ptr<SocketRWMsg> socketRWMsg) {
    int fd = socketRWMsg->fd;
    if (socketRWMsg->isRead) {
        const int BUFF_SIZE = 512;
        int len;
        char readBuff[BUFF_SIZE];
        do {
            len = read(fd, &readBuff, sizeof(readBuff));
            if (len > 0) {
                OnSocketData(fd, readBuff, len);
            }
        } while (len == BUFF_SIZE);
        // 情况一: (数据刚好 == BUFF_SIZE) 下一次读取，返回 len == -1 && errno == EAGAIN 读完数据
        // 情况二：(数据超出 BUFF_SIZE) 最后一次读取，len > 0
        if (len <= 0 && errno != EAGAIN) {
            // 情况一: (len == 0) 对端关闭
            // 情况二：(len == -1 && errno != EAGAIN) 出错
            if (Sunnet::instance->GetConn(fd)) { // 保证close只调用一次
                OnSocketClose(fd);
                Sunnet::instance->CloseConn(fd);
            } // 效率至上：大部分情况会正常read，GetConn会加锁，所以尽可能缩小这条条件判断的操作
        }
    }

    if (socketRWMsg->isWrite) {
        if (Sunnet::instance->GetConn(fd)) { // 保证可写有效，即对象存在
            OnSocketWritable(fd);
        }
    }
}

void Service::OnSocketData(int fd, const char* buff, int len) {
    cout << "OnSocketData: " << fd << " buff : " << buff << endl; 
    lua_getglobal(luaState, "OnSocketData");
    lua_pushinteger(luaState, fd);
    lua_pushlstring(luaState, buff, len);
    if (lua_pcall(luaState, 2, 0, 0) != 0) {
        cout << "call lua OnSocketData failed : " << lua_tostring(luaState, -1) << endl;
    }

    // auto writer = writers[fd];
    // writer->EntireWrite(shared_ptr<char>(writeBuff), 6);
    // writer->LingerClose();
}

void Service::OnSocketWritable(int fd) {
    cout << "OnSocketWritable: " << fd << endl;
    auto writer = writers[fd];
    writer->OnWriteable();
}

void Service::OnSocketClose(int fd) { // 关闭连接前
    writers.erase(fd);
    cout << "OnSocketClose: " << fd << endl;

    lua_getglobal(luaState, "OnSocketClose");
    lua_pushinteger(luaState, fd);
    if (lua_pcall(luaState, 1, 0, 0) != 0) {
        cout << "call lua OnSocketClose failed : " << lua_tostring(luaState, -1) << endl;
    }
}

